#include "pth_pool.h"

static void push_taskqueue(taskqueue *queue, task *curtask); //将任务加入队列
static task *take_taskqueue(taskqueue *queue);               // take_taskqueue 从任务队列头部提取任务,并在队列中删除此任务
static void init_taskqueue(taskqueue *queue);                //初始化任务队列
static void destory_taskqueue(taskqueue *queue);             //销毁任务队列

static void *thread_do(struct thread *pthread);                              /*线程运行的逻辑函数*/
static int create_thread(threadpool *pool, struct thread **pthread, int id); /*创建线程*/

static void push_taskqueue(taskqueue *queue, task *curtask) //将任务加入队列
{
    if (queue->len < MAX_TASK_NUM)
    {
        queue->rear->next = curtask;
        queue->rear = curtask;
        queue->len++;
    }
    else
    {
        printf("队列长度已抵达上限，取消本次入队\n");
    }
}

static task *take_taskqueue(taskqueue *queue) // take_taskqueue 从任务队列头部提取任务,并在队列中删除此任务
{
    if (queue->rear == queue->front)
        return nullptr;
    task *ret = queue->front->next;
    queue->front->next = ret->next;
    queue->len--;
    if (queue->rear == ret)
    {
        queue->rear = queue->front;
    }
    return ret;
}

static void init_taskqueue(taskqueue *queue) //初始化任务队列
{
    queue->has_jobs = (staconv *)malloc(sizeof(staconv));
    queue->front = queue->rear = (task *)malloc(sizeof(task));
    queue->front->next = nullptr;
    queue->len = 0;

    if (pthread_mutex_init(&queue->mutex, NULL) != 0)
    {
        printf("pool_taskqueue_mutex init fail\n");
        exit(1);
    }
    if (pthread_mutex_init(&queue->has_jobs->mutex, NULL) != 0)
    {
        printf("pool_taskqueue_status_mutex init fail\n");
        exit(1);
    }
    if (pthread_condattr_setclock(&queue->has_jobs->cattr, CLOCK_MONOTONIC) != 0)
    {
        printf("pool_taskqueue_status_condattr setclock fail\n");
        exit(1);
    }
    if (pthread_cond_init(&queue->has_jobs->cond, &queue->has_jobs->cattr) != 0)
    {
        printf("pool_taskqueue_status_cond init fail\n");
        exit(1);
    }
}

static void destory_taskqueue(taskqueue *queue) //销毁任务队列，未验证，暂勿使用
{
    while (queue->front)
    {
        queue->rear = queue->front->next;
        free(queue->front);
        queue->front = queue->rear;
    }
}

/*线程运行的逻辑函数*/
static void *thread_do(struct thread *pthread)
{
    /* 设置线程名字 */
    char thread_name[128] = {0};
    sprintf(thread_name, "thread-pool-%d", pthread->id);
    // printf("thread-pool-%d\n", pthread->id);
    /* 获得线程池*/
    threadpool *pool = pthread->pool;

    /* 在线程池初始化时,用于已经创建线程的计数,执行 pool->num_threads++ */
    pthread_mutex_lock(&pool->thcount_lock);
    pool->num_threads++;
    pthread_mutex_unlock(&pool->thcount_lock);

    /*线程一直循环往复运行,直到 pool->is_alive 变为 false*/
    while (pool->is_alive)
    {
        /*如果任务队列中还要任务,则继续运行,否则阻塞*/
        pthread_mutex_lock(&pool->queue.has_jobs->mutex);
        struct timespec tv;
        while (pool->queue.len <= 0)
        {
            clock_gettime(CLOCK_MONOTONIC, &tv);
            tv.tv_nsec += BLOCK_TIME_LIMIT; // 设置5ms后超时
            pthread_cond_timedwait(&pool->queue.has_jobs->cond, &pool->queue.has_jobs->mutex, &tv);
        }
        pthread_mutex_unlock(&pool->queue.has_jobs->mutex);

        if (pool->is_alive)
        {
            // printf("thread-pool-%d 线程正在工作\n", pthread->id);
            /*执行到此位置,表明线程在工作,需要对工作线程数量进行计数*/
            pthread_mutex_lock(&pool->thcount_lock);
            pool->num_working++;
            pthread_mutex_unlock(&pool->thcount_lock);

            /* 从任务队列的队首提取任务,并执行*/
            void (*func)(void *);
            void *arg;
            // take_taskqueue 从任务队列头部提取任务,并在队列中删除此任务
            pthread_mutex_lock(&pool->queue.mutex);
            task *curtask = take_taskqueue(&pool->queue);
            pthread_mutex_unlock(&pool->queue.mutex);
            // printf("take finish 队列剩余任务数:%d\n", pool->queue.len);
            if (curtask)
            {
                // printf("开始执行任务,现有工作线程数量:%d\n", pool->num_working);
                func = curtask->function;
                arg = curtask->arg;
                //执行任务
                (*func)(arg);
                //释放任务,因除初始化其它指针均未使用malloc分配空间，故不需要free
                // free(curtask);
            }
            /*执行到此位置,表明线程已经将任务执行完成,需更改工作线程数量*/
            // printf("任务完成,队列剩余任务数:%d\n", pool->queue.len);
            pthread_mutex_lock(&pool->thcount_lock);
            pool->num_working--;
            pthread_mutex_unlock(&pool->thcount_lock);
            if (pool->num_working == 0)
            {
            }
        }
    }
    /*运行到此位置表明,线程将要退出,需更改当前线程池中的线程数量*/
    pthread_mutex_lock(&pool->thcount_lock);
    pool->num_threads--;
    pthread_mutex_unlock(&pool->thcount_lock);

    return NULL;
}

/*创建线程*/
static int create_thread(threadpool *pool, struct thread **pthread, int id)
{
    //为 thread 分配内存空间
    *pthread = (struct thread *)malloc(sizeof(struct thread));
    if (pthread == NULL)
    {
        perror("creat_thread(): Could not allocate memory for thread\n");
        return -1;
    }
    //设置这个 thread 的属性
    (*pthread)->pool = pool;
    (*pthread)->id = id;
    //创建线程
    pthread_create(&(*pthread)->pthread, NULL, (void *(*)(void *))thread_do, (*pthread));
    //初始化线程属性,为分离状态
    pthread_detach((*pthread)->pthread);
    return 0;
}

/*线程池初始化函数*/
struct threadpool *initTheadPool(int num_threads)
{
    //创建线程池空间
    threadpool *pool;
    pool = (threadpool *)malloc(sizeof(struct threadpool));
    pool->num_threads = 0;
    pool->num_working = 0;
    //初始化互斥量和条件变量
    if (pthread_mutex_init(&pool->thcount_lock, NULL) != 0)
    {
        printf("pool_mutex thcount_lock init fail\n");
        exit(1);
    }
    if (pthread_cond_init(&pool->threads_all_idle, NULL) != 0)
    {
        printf("pool_cond threads_all_idle init fail\n");
        exit(1);
    }
    //初始化任务队列
    init_taskqueue(&pool->queue);
    printf("taskqueue init success\n");
    //创建可容纳 num_threads 个的线程数组
    pool->threads = (struct thread **)malloc(num_threads * sizeof(struct thread *));
    pool->is_alive = TRUE;

    //创建线程
    for (int i = 0; i < num_threads; ++i)
    {
        create_thread(pool, &pool->threads[i], i); // i 为线程 id,
    }
    //等等所有的线程创建完毕,在每个线程运行函数中将进行 pool->num_threads++操作
    //因此,此处为忙等待,直到所有的线程创建完毕,并?上运行阻塞代码时才返回。
    while (pool->num_threads != num_threads)
    {
    }
    printf("thread pool init success\n");
    return pool;
}

/*向线程池中添加任务*/
void addTask2ThreadPool(threadpool *pool, task *curtask)
{
    //将任务加入队列
    pthread_mutex_lock(&pool->queue.mutex);
    push_taskqueue(&pool->queue, curtask);
    pthread_mutex_unlock(&pool->queue.mutex);
}
